Skip to content

LangGraph 用户需求澄清与研究规划详细解读

📚 概述

本文档详细解读 Scoping(需求澄清与研究规划) 在 Deep Research 系统中的关键作用。这是整个研究流程的第一步,也是最重要的一步——如果需求理解错误,后续所有研究都将偏离方向。

核心问题: 用户的初始请求往往模糊、不完整,缺少关键信息。

示例:

  • ❌ "研究最好的咖啡店" → 最好指什么?咖啡质量?氛围?价格?
  • ❌ "对比 A 和 B" → 对比哪些维度?技术?价格?用户体验?
  • ❌ "分析特斯拉" → 分析什么?股票?技术?市场?

Scoping 的目标: 通过智能对话,将模糊请求转化为清晰、结构化的研究简报。


📚 术语表

术语名称LangGraph 定义和解读Python 定义和说明重要程度
Scoping需求澄清与研究规划阶段,通过对话明确用户真实需求并生成研究简报N/A (工作流概念)⭐⭐⭐⭐⭐
ClarifyWithUser结构化输出 Schema,判断是否需要向用户提问class ClarifyWithUser(BaseModel)⭐⭐⭐⭐⭐
ResearchQuestion结构化输出 Schema,用于生成研究简报class ResearchQuestion(BaseModel)⭐⭐⭐⭐⭐
CommandLangGraph 控制流对象,用于动态路由和状态更新Command(goto="node", update={...})⭐⭐⭐⭐⭐
Structured Output强制 LLM 输出符合预定义 Schema 的技术model.with_structured_output(Schema)⭐⭐⭐⭐⭐
get_buffer_string将消息历史转化为字符串格式from langchain_core.messages import get_buffer_string⭐⭐⭐⭐
MessagesStateLangGraph 预定义状态类,管理消息历史class MessagesState(TypedDict)⭐⭐⭐⭐
Research Brief详细的研究规划文档,指导后续研究方向字符串,包含研究主题、范围、标准等⭐⭐⭐⭐⭐
LLM-as-judge使用 LLM 评估其他 LLM 输出质量的技术评估函数,调用 LLM 并返回评分⭐⭐⭐⭐

🎯 核心概念

什么是 Scoping?

Scoping 是 Deep Research 系统的第一阶段,包含两个子步骤:

  1. User Clarification(用户澄清) - 判断是否需要向用户提问以澄清需求
  2. Brief Generation(简报生成) - 将对话历史转化为详细的研究简报

完整流程:

用户输入: "研究最好的咖啡店"

┌─────────────────────────────────────┐
│ Step 1: 用户澄清                     │
│ LLM 分析: "最好"的标准不明确          │
│ 生成问题: "您关注咖啡质量、氛围还是价格?"│
└─────────────────────────────────────┘

用户回复: "咖啡质量"

┌─────────────────────────────────────┐
│ Step 2: 简报生成                     │
│ 将对话转化为研究简报:                 │
│ "研究旧金山地区以咖啡质量著称的咖啡店,│
│  重点关注专业评分、豆源、烘焙技术..."  │
└─────────────────────────────────────┘

为什么需要 Scoping?

问题 1:用户请求缺少关键信息

常见缺失信息:

  • 范围和边界 - 应该包括/排除什么?
  • 受众和目的 - 这个研究是给谁看的,为什么需要?
  • 具体要求 - 有特定的来源、时间范围、约束吗?
  • 术语澄清 - 领域术语或缩写是什么意思?

示例对比:

模糊请求缺失信息澄清问题
"研究最好的咖啡店"地点、标准"哪个城市?关注什么标准?"
"对比 OpenAI 和 Anthropic"对比维度"对比技术能力、价格还是使用体验?"
"分析 TSLA"分析角度"分析股票表现、技术创新还是市场竞争?"

问题 2:避免错误假设

如果不澄清需求,LLM 可能基于自身偏好做出假设:

python
# ❌ 错误假设示例
User: "研究最好的咖啡店"
LLM (假设): "用户可能关心价格和便利性"
→ 研究结果: 连锁咖啡店(Starbucks, Peet's)

# ✅ 正确流程
User: "研究最好的咖啡店"
LLM: "您关注咖啡质量、氛围还是价格?"
User: "咖啡质量"
→ 研究结果: 精品咖啡店(Blue Bottle, Sightglass)

🔧 技术实现详解

1. 状态定义

python
from typing_extensions import Optional, Annotated, Sequence
from langchain_core.messages import BaseMessage
from langgraph.graph import MessagesState
from langgraph.graph.message import add_messages
from pydantic import BaseModel, Field

# ===== 状态定义 =====

class AgentState(MessagesState):
    """
    主状态类,继承自 MessagesState

    MessagesState 提供:
    - messages: 消息历史(自动使用 add_messages reducer)

    新增字段:
    - research_brief: 生成的研究简报
    - supervisor_messages: 与 supervisor 的通信
    """
    research_brief: Optional[str]
    supervisor_messages: Annotated[Sequence[BaseMessage], add_messages]

关键点:

  • MessagesState - LangGraph 预定义基类,自动管理消息历史
  • add_messages - Reducer,自动处理消息追加/更新/删除
  • Optional[str] - 研究简报初始为 None,生成后才有值

2. Structured Output Schema

为什么使用 Structured Output?

传统方式(字符串输出)的问题:

python
# ❌ 不可靠的字符串解析
response = llm.invoke("判断是否需要澄清,回答 yes/no")
# 可能输出: "Yes, I think..."  "YES"  "yes."  "需要澄清"
# → 难以可靠解析

Structured Output 的优势:

python
# ✅ 可靠的结构化输出
class ClarifyWithUser(BaseModel):
    need_clarification: bool  # 强制 bool 类型
    question: str
    verification: str

response = llm.with_structured_output(ClarifyWithUser).invoke(...)
# 保证返回: response.need_clarification 是 True/False

完整 Schema 定义:

python
class ClarifyWithUser(BaseModel):
    """用户澄清决策和问题 Schema"""

    need_clarification: bool = Field(
        description="是否需要向用户提问以澄清需求",
    )
    question: str = Field(
        description="如果需要澄清,向用户提出的问题(使用 Markdown 格式)",
    )
    verification: str = Field(
        description="如果不需要澄清,确认开始研究的消息",
    )

工作流程:

python
if response.need_clarification:
    # 返回问题给用户
    return {"messages": [AIMessage(content=response.question)]}
else:
    # 确认开始研究
    return {"messages": [AIMessage(content=response.verification)]}

3. Prompt 设计

核心 Prompt(clarify_with_user_instructions):

python
clarify_with_user_instructions = """
这是用户与你的对话历史:
<Messages>
{messages}
</Messages>

今天的日期是 {date}

评估是否需要提问澄清,或者用户已经提供了足够信息。

重要提示: 如果你已经在对话历史中问过澄清问题,几乎总是不需要再问。
只在绝对必要时才提出新问题。

如果有缩写、简称或未知术语,请用户澄清。

如果需要提问,遵循以下准则:
- 简洁的同时收集所有必要信息
- 确保收集执行研究任务所需的全部信息
- 适当时使用项目符号或编号列表以提高清晰度
- 确保使用 Markdown 格式,可被 Markdown 渲染器正确显示
- 不要询问不必要的信息或用户已经提供的信息

以有效 JSON 格式响应,包含以下键:
"need_clarification": boolean,
"question": "<向用户提问以澄清研究范围>",
"verification": "<确认我们将开始研究的消息>"

如果需要澄清问题,返回:
"need_clarification": true,
"question": "<你的澄清问题>",
"verification": ""

如果不需要澄清问题,返回:
"need_clarification": false,
"question": "",
"verification": "<确认消息:你将基于提供的信息开始研究>"

对于不需要澄清时的确认消息:
- 确认你有足够信息继续
- 简要总结你从请求中理解的关键内容
- 确认你现在将开始研究过程
- 保持简洁和专业
"""

Prompt 设计原则:

  1. 防止重复提问

    重要提示: 如果你已经在对话历史中问过澄清问题,几乎总是不需要再问。
  2. 明确输出格式

    • 使用 JSON Schema
    • 提供正负示例(需要澄清 vs 不需要澄清)
  3. 提供上下文

    • 包含完整对话历史
    • 提供当前日期(对时间敏感的研究很重要)
  4. 质量要求

    • 使用 Markdown 格式
    • 简洁但完整
    • 项目符号提高可读性

4. 节点实现:clarify_with_user

python
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, AIMessage, get_buffer_string
from langgraph.types import Command

# 初始化模型
model = init_chat_model(model="openai:gpt-4.1", temperature=0.0)

def clarify_with_user(state: AgentState) -> Command:
    """
    判断是否需要向用户提问以澄清需求

    使用 Structured Output 确保可靠的决策

    Returns:
        Command 对象,指向下一个节点:
        - 如果需要澄清 → goto=END (返回问题给用户)
        - 如果不需要澄清 → goto="write_research_brief"
    """
    # 设置 Structured Output 模型
    structured_output_model = model.with_structured_output(ClarifyWithUser)

    # 调用模型进行判断
    response = structured_output_model.invoke([
        HumanMessage(content=clarify_with_user_instructions.format(
            messages=get_buffer_string(messages=state["messages"]),
            date=get_today_str()
        ))
    ])

    # 基于判断结果路由
    if response.need_clarification:
        # 需要澄清:结束当前执行,返回问题给用户
        return Command(
            goto=END,
            update={"messages": [AIMessage(content=response.question)]}
        )
    else:
        # 不需要澄清:继续到简报生成
        return Command(
            goto="write_research_brief",
            update={"messages": [AIMessage(content=response.verification)]}
        )

关键技术:Command 对象

python
Command(
    goto="next_node",  # 下一个要执行的节点(或 END)
    update={...}        # 要更新的状态字段
)

Command 的优势:

  • 🎯 灵活路由 - 节点可以动态决定下一步
  • 🔄 状态更新 - 同时更新状态和跳转
  • 📊 清晰语义 - 代码意图一目了然

对比传统条件边:

python
# ❌ 传统方式:需要单独的路由函数
def should_continue(state):
    if needs_clarification(state):
        return "end"
    return "write_brief"

builder.add_conditional_edges("clarify", should_continue, {...})

# ✅ 使用 Command:决策和路由在一个节点中
def clarify_with_user(state):
    if response.need_clarification:
        return Command(goto=END, update={...})
    return Command(goto="write_research_brief", update={...})

5. 节点实现:write_research_brief

python
class ResearchQuestion(BaseModel):
    """研究简报 Schema"""
    research_brief: str = Field(
        description="详细的研究简报,将指导后续研究",
    )

def write_research_brief(state: AgentState):
    """
    将对话历史转化为详细的研究简报

    使用 Structured Output 确保简报格式符合要求
    """
    # 设置 Structured Output 模型
    structured_output_model = model.with_structured_output(ResearchQuestion)

    # 生成研究简报
    response = structured_output_model.invoke([
        HumanMessage(content=transform_messages_into_research_topic_prompt.format(
            messages=get_buffer_string(state.get("messages", [])),
            date=get_today_str()
        ))
    ])

    # 更新状态
    return {
        "research_brief": response.research_brief,
        "supervisor_messages": [HumanMessage(content=f"{response.research_brief}.")]
    }

研究简报的质量要求:

好的研究简报应该包含:

  1. 明确的研究主题 - 要研究什么?
  2. 具体的范围 - 包括/排除什么?
  3. 评估标准 - 如何判断"好"?
  4. 优先信息源 - 首选什么类型的来源?
  5. 时间要求 - 截止到什么时候的信息?

示例对比:

质量研究简报内容
❌ 差"研究旧金山的咖啡店"
⚠️ 一般"研究旧金山的精品咖啡店,关注咖啡质量"
✅ 好"研究旧金山以咖啡质量著称的精品咖啡店。重点关注豆源、烘焙技术、专业评分(如 Coffee Review)和用户评价。优先使用咖啡店官网、第三方专业评测(Specialty Coffee Association)和评价聚合网站(Yelp, Google Reviews)。截止到 2025年7月的最新数据。"

6. 图构建

python
from langgraph.graph import StateGraph, START, END

# 构建 Scoping 工作流
scope_builder = StateGraph(AgentState, input_schema=AgentInputState)

# 添加节点
scope_builder.add_node("clarify_with_user", clarify_with_user)
scope_builder.add_node("write_research_brief", write_research_brief)

# 添加边
scope_builder.add_edge(START, "clarify_with_user")
scope_builder.add_edge("write_research_brief", END)

# 编译
scope_research = scope_builder.compile()

# 🎨 可视化图结构
from IPython.display import Image, display
display(Image(scope_research.get_graph().draw_mermaid_png()))

图的执行流程:

START

clarify_with_user
  ↓ (decision)
  ├─ need_clarification? → END (返回问题)
  └─ sufficient info? → write_research_brief → END

🎭 实战案例:咖啡店研究

示例 1:需要澄清

python
from langchain_core.messages import HumanMessage
from langgraph.checkpoint.memory import MemorySaver

# 编译带 checkpointer 的图(用于测试)
checkpointer = MemorySaver()
scope = scope_builder.compile(checkpointer=checkpointer)

# 第一次调用:模糊请求
thread = {"configurable": {"thread_id": "1"}}
result = scope.invoke({
    "messages": [HumanMessage(content="我想研究旧金山最好的咖啡店。")]
}, config=thread)

print(result['messages'][-1].content)

输出:

您能具体说明什么标准对您来说最重要吗?例如:
- 咖啡质量
- 氛围和环境
- Wi-Fi 可用性
- 食物选项
- 其他因素?

第二次调用:提供澄清

python
result = scope.invoke({
    "messages": [HumanMessage(content="咖啡质量是最重要的标准。")]
}, config=thread)

print(result['messages'][-1].content)

输出:

感谢澄清,咖啡质量是您的首要标准。我有足够信息继续,
现在将开始研究旧金山基于咖啡质量的最佳咖啡店。

生成的研究简报:

python
print(result["research_brief"])

输出:

我想识别和评估旧金山被认为是基于咖啡质量的最佳咖啡店。
我的研究应专注于分析和比较旧金山地区的咖啡店,以咖啡质量
作为主要标准。我对评估咖啡质量的方法持开放态度(例如,
专家评测、客户评分、精品咖啡认证),对氛围、地点、Wi-Fi
或食物选项没有约束,除非它们直接影响感知的咖啡质量。

请优先考虑主要来源,如咖啡店官网、信誉良好的第三方咖啡
评测组织(如 Coffee Review 或 Specialty Coffee Association)
以及知名评价聚合网站(如 Google 或 Yelp),在这些网站可以
找到关于咖啡质量的直接客户反馈。

研究应产生一个有充分支持的列表或排名,强调截止到 2025年7月
的最新可用数据所显示的咖啡质量。

示例 2:立即满足(无需澄清)

python
thread2 = {"configurable": {"thread_id": "2"}}
result = scope.invoke({
    "messages": [HumanMessage(content="""
    我想研究旧金山基于咖啡质量的最佳咖啡店。
    请关注专业评分(Coffee Review)、豆源和烘焙技术。
    优先使用官网和专业评测网站。
    """)]
}, config=thread2)

print(result['messages'][-1].content)

输出:

感谢您提供具体的研究参数。您已经明确指定了地点(旧金山)、
主要标准(咖啡质量)以及评估方法(专业评分、豆源、烘焙技术)。
我现在将开始基于提供的标准进行研究。

直接生成简报,无需额外澄清。


🎓 核心知识点总结

LangGraph 特有概念

1. Structured Output 的强大之处

传统字符串输出的问题:

python
# ❌ 不可靠
response = llm.invoke("判断是否需要澄清,回答 true/false")
# 可能返回: "True"  "true"  "TRUE"  "yes"  "I think true"
if "true" in response.lower():  # 脆弱的解析
    ...

Structured Output 的优势:

python
# ✅ 可靠
class Decision(BaseModel):
    need_clarification: bool

response = llm.with_structured_output(Decision).invoke(...)
if response.need_clarification:  # 类型安全
    ...

重要提示: Structured Output 使用 LLM 的 function calling 能力,确保输出符合 Schema。


2. Command 对象的灵活性

Command vs 条件边对比:

特性条件边Command
路由决策单独的路由函数节点内部决策
状态更新分离的逻辑一起处理
代码组织分散在多处集中在节点中
灵活性较低

使用建议:

  • 简单静态路由 - 使用条件边
  • 复杂动态路由 - 使用 Command

3. MessagesState 的便利性

手动管理 vs MessagesState:

python
# ❌ 手动管理消息
class MyState(TypedDict):
    messages: list

def node(state):
    new_messages = state["messages"] + [new_msg]  # 手动追加
    return {"messages": new_messages}

# ✅ 使用 MessagesState
class MyState(MessagesState):
    pass  # 自动包含 messages + add_messages reducer

def node(state):
    return {"messages": [new_msg]}  # 自动追加

add_messages Reducer 功能:

  • 🔄 追加消息 - 默认行为
  • 🔄 覆盖消息 - 如果提供相同 ID
  • 🔄 删除消息 - 使用 RemoveMessage

Python 特有知识点

1. Pydantic BaseModel

python
from pydantic import BaseModel, Field

class ClarifyWithUser(BaseModel):
    need_clarification: bool = Field(
        description="是否需要澄清"  # 这个描述会传递给 LLM
    )

Field 的作用:

  • 📝 description - 告诉 LLM 这个字段的用途
  • 📝 examples - 提供示例值
  • 📝 constraints - 添加验证规则

2. get_buffer_string 工具函数

python
from langchain_core.messages import get_buffer_string

messages = [
    HumanMessage("你好"),
    AIMessage("你好!有什么可以帮您?"),
    HumanMessage("告诉我关于咖啡的信息")
]

buffer = get_buffer_string(messages)
# 输出:
# Human: 你好
# AI: 你好!有什么可以帮您?
# Human: 告诉我关于咖啡的信息

用途: 将消息列表格式化为 LLM 可读的字符串。


💡 最佳实践

1. 何时需要澄清?

需要澄清的场景:

  • ✅ 请求包含模糊术语("最好"、"顶级"、"优秀")
  • ✅ 缺少关键参数(地点、时间、标准)
  • ✅ 使用缩写或专业术语
  • ✅ 对比任务但未指定维度

不需要澄清的场景:

  • ❌ 请求非常具体("2024年Q4 TSLA 股票价格走势")
  • ❌ 已经在对话中澄清过
  • ❌ 常识性请求("Python 如何定义函数")

2. Prompt 设计技巧

技巧 1:提供正负示例

python
prompt = """
如果需要澄清,返回:
{
  "need_clarification": true,
  "question": "您关注的是咖啡质量、氛围还是价格?",
  "verification": ""
}

如果不需要澄清,返回:
{
  "need_clarification": false,
  "question": "",
  "verification": "我将基于您提供的标准开始研究。"
}
"""

技巧 2:防止重复提问

python
# 在 Prompt 中明确指出
"如果你已经在对话历史中问过澄清问题,几乎总是不需要再问。"

技巧 3:使用 XML 标签组织上下文

python
prompt = """
<Messages>
{messages}
</Messages>

<Instructions>
...
</Instructions>

<Output Format>
...
</Output Format>
"""

3. 研究简报的质量标准

评估标准:

标准说明示例
完整性包含所有用户提到的要点如果用户提到"咖啡质量",简报必须包含
无假设不添加用户未提及的偏好用户未提到"价格",不应假设价格重要
具体性明确的评估标准和来源"专业评分(Coffee Review)"而非"好的评价"
可执行提供足够细节供研究 Agent 执行明确时间范围、地理范围、数据源

4. 错误处理

常见问题:

问题 1:LLM 拒绝使用 Structured Output

python
# ❌ LLM 返回自然语言而非 JSON
try:
    response = llm.with_structured_output(ClarifyWithUser).invoke(...)
except Exception as e:
    # 降级策略:使用正则表达式解析
    ...

解决方案:

  • 使用更强的模型(GPT-4 而非 GPT-3.5)
  • 在 Prompt 中强调输出格式要求
  • 提供更多示例

问题 2:对话历史太长

python
# 如果对话超过 100 轮
if len(state["messages"]) > 100:
    # 只保留最近 50 轮
    recent_messages = state["messages"][-50:]
    buffer = get_buffer_string(recent_messages)

🔍 评估方法

设计评估器

评估目标:

  1. 研究简报是否包含所有用户提到的标准?
  2. 研究简报是否避免了用户未提及的假设?

评估器 1:Success Criteria(成功标准)

python
from pydantic import BaseModel, Field

class Criteria(BaseModel):
    criteria_text: str = Field(description="具体的成功标准")
    reasoning: str = Field(description="评估推理")
    is_captured: bool = Field(description="是否在简报中体现")

def evaluate_success_criteria(outputs: dict, reference_outputs: dict):
    """
    评估研究简报是否包含所有必需标准
    """
    research_brief = outputs["research_brief"]
    success_criteria = reference_outputs["criteria"]

    model = ChatOpenAI(model="gpt-4.1", temperature=0)
    structured_model = model.with_structured_output(Criteria)

    # 批量评估每个标准
    individual_evaluations = []
    for criterion in success_criteria:
        response = structured_model.invoke([
            HumanMessage(content=f"""
            研究简报: {research_brief}

            评估标准: {criterion}

            判断这个标准是否在研究简报中得到充分体现。
            """)
        ])
        individual_evaluations.append(response)

    # 计算总分
    captured_count = sum(1 for eval in individual_evaluations if eval.is_captured)
    total_count = len(individual_evaluations)

    return {
        "key": "success_criteria_score",
        "score": captured_count / total_count,
        "individual_evaluations": individual_evaluations
    }

评估器 2:No Assumptions(无假设)

python
class NoAssumptions(BaseModel):
    no_assumptions: bool = Field(
        description="研究简报是否避免了未经用户明确的假设"
    )
    reasoning: str = Field(description="评估推理")

def evaluate_no_assumptions(outputs: dict, reference_outputs: dict):
    """
    评估研究简报是否避免了错误假设
    """
    research_brief = outputs["research_brief"]
    success_criteria = reference_outputs["criteria"]

    model = ChatOpenAI(model="gpt-4.1", temperature=0)
    structured_model = model.with_structured_output(NoAssumptions)

    response = structured_model.invoke([
        HumanMessage(content=f"""
        研究简报: {research_brief}
        用户明确的标准: {success_criteria}

        评估研究简报是否包含了用户未明确提及的假设或偏好。
        """)
    ])

    return {
        "key": "no_assumptions_score",
        "score": response.no_assumptions,
        "reasoning": response.reasoning
    }

运行评估:

python
from langsmith import Client

client = Client()

# 创建数据集
dataset_name = "scoping_quality"
dataset = client.create_dataset(dataset_name, description="Scoping 质量评估")

# 添加测试用例
client.create_examples(
    dataset_id=dataset.id,
    examples=[
        {
            "inputs": {"messages": conversation_1},
            "outputs": {"criteria": ["当前年龄 25", "目标退休年龄 45", "高风险承受", ...]}
        }
    ]
)

# 运行评估
client.evaluate(
    lambda inputs: scope.invoke(inputs),
    data=dataset_name,
    evaluators=[evaluate_success_criteria, evaluate_no_assumptions],
    experiment_prefix="Scoping Quality"
)

📖 扩展阅读


🎉 总结

Scoping 是 Deep Research 的关键第一步:

  1. 智能澄清 - 使用 Structured Output 可靠判断是否需要提问
  2. Command 控制流 - 灵活路由到不同节点
  3. 研究简报生成 - 将对话转化为结构化规划
  4. 评估驱动 - 使用 LLM-as-judge 持续改进质量

核心技巧:

  • ClarifyWithUser Schema 确保可靠决策
  • Command 对象实现动态路由
  • get_buffer_string 格式化消息历史
  • 评估器验证简报质量

通过 Scoping,我们避免了基于模糊需求进行研究的陷阱,为后续的深度研究打下坚实基础!

🎯 下一步: 让我们继续到 9.2 研究智能体基础 — 学习如何构建能自主搜索、反思、决策的研究 Agent!

基于 MIT 许可证发布。内容版权归作者所有。